package drds.global_transaction.transaction_coordinator.transaction_coordinator;

import drds.global_transaction.api.api.BranchTransactionStatus;
import drds.global_transaction.api.api.GlobalTransactionStatus;
import drds.global_transaction.api.api.ResourceManagerInbound;
import drds.global_transaction.api.exception.GlobalTransactionException;
import drds.global_transaction.api.exception.TransactionExceptionCode;
import drds.global_transaction.common.GlobalTransactionIds;
import drds.global_transaction.transaction_coordinator.TransactionIds;
import drds.global_transaction.transaction_coordinator.lock_manager.LockManager;
import drds.global_transaction.transaction_coordinator.lock_manager.LockManagerFactory;
import drds.global_transaction.transaction_coordinator.transaction_session.BranchTransactionSession;
import drds.global_transaction.transaction_coordinator.transaction_session.GlobalTransactionSession;
import drds.global_transaction.transaction_coordinator.transaction_session.TransactionSessionManagerFactory;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class GlobalTransactionManagersImpl implements GlobalTransactionManagers
{
	private LockManager lockManager = LockManagerFactory.get();
	@Setter
	@Getter
	private ResourceManagerInbound resourceManagerInbound;

	@Override
	public String globalTransactionBegin(String applicationId, String transactionServiceGroup, String transactionName, int timeout) throws GlobalTransactionException
	{
		GlobalTransactionSession globalTransactionSession = new GlobalTransactionSession(applicationId, transactionServiceGroup, transactionName, timeout);
		globalTransactionSession.addSessionLifecycleListener(TransactionSessionManagerFactory.getRootTransactionSessionManager());
		globalTransactionSession.begin();
		return GlobalTransactionIds.generateGlobalTransactionIdWithAddressAndPort(globalTransactionSession.getGlobalTransactionId());
	}

	@Override
	public GlobalTransactionStatus globalTransactionCommit(String globalTransactionIdWithAddressAndPort) throws GlobalTransactionException
	{
		GlobalTransactionSession globalTransactionSession = TransactionSessionManagerFactory.findGlobalTransactionSession(GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort));
		if (globalTransactionSession == null)
		{
			return GlobalTransactionStatus.finished;
		}
		GlobalTransactionStatus globalTransactionStatus = globalTransactionSession.getGlobalTransactionStatus();

		globalTransactionSession.closeAndClean(); // Highlight: Firstly, close the transaction_session, then no more
		// branch can be registered.

		if (globalTransactionStatus == GlobalTransactionStatus.begin)
		{
			if (globalTransactionSession.canBeAsynchronousCommitted())
			{
				asynchronousGlobalTransactionCommit(globalTransactionSession);
			} else
			{
				doGlobalTransactionCommit(globalTransactionSession, false);
			}
		}
		return globalTransactionSession.getGlobalTransactionStatus();
	}

	private void asynchronousGlobalTransactionCommit(GlobalTransactionSession globalTransactionSession) throws GlobalTransactionException
	{
		globalTransactionSession.addSessionLifecycleListener(TransactionSessionManagerFactory.getAsynchronousTransactionSessionManager());
		TransactionSessionManagerFactory.getAsynchronousTransactionSessionManager().addGlobalTransactionSession(globalTransactionSession);
	}

	@Override
	public void doGlobalTransactionCommit(GlobalTransactionSession globalTransactionSession, boolean retrying) throws GlobalTransactionException
	{
		// 可能需要先整体进行判断是否异常,再进行分支事务的操作
		for (BranchTransactionSession branchTransactionSession : globalTransactionSession.getSortedBranchTransactionSessionList())
		{
			if (branchTransactionSession.getBranchTransactionStatus() == BranchTransactionStatus.phase_one_failed)
			{
				continue;
			}
			try
			{
				BranchTransactionStatus branchTransactionStatus = resourceManagerInbound.branchTransactionCommit(GlobalTransactionIds.generateGlobalTransactionIdWithAddressAndPort(branchTransactionSession.getGlobalTransactionId()), branchTransactionSession.getBranchTransactionId(), branchTransactionSession.getResourceId(), branchTransactionSession.getApplicationData());
				switch (branchTransactionStatus)
				{
				case phase_two_committed:
					globalTransactionSession.unlockAndRemoveBranchTransactionSession(branchTransactionSession);
					continue;
				case phase_two_commit_failed_and_unretriable:
					if (globalTransactionSession.canBeAsynchronousCommitted())
					{
						log.error("By [" + branchTransactionStatus + "], failed to globalTransactionCommit branch" + " " + branchTransactionSession);
						continue;
					} else
					{
						globalTransactionSession.changeGlobalTransactionStatus(GlobalTransactionStatus.commit_failed);
						globalTransactionSession.end();
						log.error("Finally, failed to globalTransactionCommit global[" + globalTransactionSession.getGlobalTransactionId() + "] since branch[" + branchTransactionSession.getBranchTransactionId() + "] globalTransactionCommit failed");
						return;
					}
				default:
					if (!retrying)
					{
						queueToRetryCommit(globalTransactionSession);
						return;
					}
					if (globalTransactionSession.canBeAsynchronousCommitted())
					{
						log.error("By [" + branchTransactionStatus + "], failed to globalTransactionCommit branch" + " " + branchTransactionSession);
						continue;
					} else
					{
						log.error("failed to globalTransactionCommit global[" + globalTransactionSession.getGlobalTransactionId() + "] since branch[" + branchTransactionSession.getBranchTransactionId() + "] globalTransactionCommit failed, will retry later.");
						return;
					}

				}

			} catch (Exception ex)
			{
				log.info("Exception committing branch " + branchTransactionSession, ex);
				if (!retrying)
				{
					queueToRetryCommit(globalTransactionSession);
					if (ex instanceof GlobalTransactionException)
					{
						throw (GlobalTransactionException) ex;
					} else
					{
						throw new GlobalTransactionException(ex);
					}
				}

			}

		}
		if (globalTransactionSession.hasBranchTransaction())
		{
			log.info("Global[" + globalTransactionSession.getGlobalTransactionId() + "] committing is NOT done.");
			return;
		}
		globalTransactionSession.changeGlobalTransactionStatus(GlobalTransactionStatus.committed);
		globalTransactionSession.end();
		log.info("Global[" + globalTransactionSession.getGlobalTransactionId() + "] committing is successfully done.");
	}
	//

	@Override
	public GlobalTransactionStatus globalTransactionRollback(String globalTransactionIdWithAddressAndPort) throws GlobalTransactionException
	{
		GlobalTransactionSession globalTransactionSession = TransactionSessionManagerFactory.findGlobalTransactionSession(GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort));
		if (globalTransactionSession == null)
		{
			return GlobalTransactionStatus.finished;
		}
		GlobalTransactionStatus globalTransactionStatus = globalTransactionSession.getGlobalTransactionStatus();
		globalTransactionSession.close(); // Highlight: Firstly, close the transaction_session, then no more branch
		// can be registered.
		if (globalTransactionStatus == GlobalTransactionStatus.begin)
		{
			globalTransactionSession.changeGlobalTransactionStatus(GlobalTransactionStatus.rollbacking);
			doGlobalTransactionRollback(globalTransactionSession, false);
		}
		return globalTransactionSession.getGlobalTransactionStatus();
	}

	@Override
	public void doGlobalTransactionRollback(GlobalTransactionSession globalTransactionSession, boolean retrying) throws GlobalTransactionException
	{
		for (BranchTransactionSession branchTransactionSession : globalTransactionSession.getReverseSortedBranchTransactionSessionList())
		{
			BranchTransactionStatus currentBranchTransactionStatus = branchTransactionSession.getBranchTransactionStatus();
			if (currentBranchTransactionStatus == BranchTransactionStatus.phase_one_failed)
			{
				continue;// 必须保证分支事务有序,出现异常的分支事务之前的所有正常的分支事务都需要回滚。如果是业务造成的回滚则需要依次逆向顺序进行回滚。
			}
			try
			{
				BranchTransactionStatus branchTransactionStatus = resourceManagerInbound.branchTransactionRollback(GlobalTransactionIds.generateGlobalTransactionIdWithAddressAndPort(branchTransactionSession.getGlobalTransactionId()), branchTransactionSession.getBranchTransactionId(), branchTransactionSession.getResourceId(), branchTransactionSession.getApplicationData());
				switch (branchTransactionStatus)
				{
				case phase_two_rollbacked:
					globalTransactionSession.unlockAndRemoveBranchTransactionSession(branchTransactionSession);
					log.error("Successfully rolled back branch " + branchTransactionSession);
					continue;
				case phase_two_rollback_failed_and_unretriable:
					GlobalTransactionStatus currentStatus = globalTransactionSession.getGlobalTransactionStatus();
					if (currentStatus.name().startsWith("Timeout"))
					{
						globalTransactionSession.changeGlobalTransactionStatus(GlobalTransactionStatus.timeout_rollback_failed);
					} else
					{
						globalTransactionSession.changeGlobalTransactionStatus(GlobalTransactionStatus.rollback_failed);
					}
					globalTransactionSession.end();// 关键是这应该如何处理
					log.error("failed to globalTransactionRollback global[" + globalTransactionSession.getGlobalTransactionId() + "] since branch[" + branchTransactionSession.getBranchTransactionId() + "] globalTransactionRollback failed");
					return;
				default:
					log.info("failed to globalTransactionRollback branch " + branchTransactionSession);
					if (!retrying)
					{
						queueToRetryRollback(globalTransactionSession);
					}
					return;

				}
			} catch (Exception ex)
			{
				log.info("Exception rollbacking branch " + branchTransactionSession, ex);
				if (!retrying)
				{
					queueToRetryRollback(globalTransactionSession);
					if (ex instanceof GlobalTransactionException)
					{
						throw (GlobalTransactionException) ex;
					} else
					{
						throw new GlobalTransactionException(ex);
					}
				}

			}

		}
		GlobalTransactionStatus globalTransactionStatus = globalTransactionSession.getGlobalTransactionStatus();
		if (globalTransactionStatus.name().startsWith("Timeout"))
		{
			globalTransactionSession.changeGlobalTransactionStatus(GlobalTransactionStatus.timeout_rollbacked);
		} else
		{
			globalTransactionSession.changeGlobalTransactionStatus(GlobalTransactionStatus.rollbacked);
		}
		globalTransactionSession.end();
	}

	//
	@Override
	public boolean lockRecordIdsCheck(String resourceId, String globalTransactionIdWithAddressAndPort, String tableNameAndLockRecordIdsStringsString) throws GlobalTransactionException
	{

		return lockManager.isLockable(resourceId, GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort), tableNameAndLockRecordIdsStringsString);

	}

	private GlobalTransactionSession checkGlobalTransactionSessionOk(long globalTransactionId, GlobalTransactionStatus globalTransactionStatus) throws GlobalTransactionException
	{
		GlobalTransactionSession globalTransactionSession = TransactionSessionManagerFactory.findGlobalTransactionSession(globalTransactionId);
		if (globalTransactionSession == null)
		{
			throw new GlobalTransactionException(TransactionExceptionCode.global_transaction_not_exist, "" + globalTransactionId + "");
		}
		if (!globalTransactionSession.isActive())
		{
			throw new GlobalTransactionException(TransactionExceptionCode.global_transaction_not_active, "Current " + "Status: " + globalTransactionSession.getGlobalTransactionStatus());
		}
		if (globalTransactionSession.getGlobalTransactionStatus() != globalTransactionStatus)
		{
			throw new GlobalTransactionException(TransactionExceptionCode.global_transaction_status_invalid, globalTransactionSession.getGlobalTransactionStatus() + " while expecting " + globalTransactionStatus);
		}
		return globalTransactionSession;
	}

	@Override
	public GlobalTransactionStatus getGlobalTransactionStatus(String globalTransactionIdWithAddressAndPort) throws GlobalTransactionException
	{
		GlobalTransactionSession globalTransactionSession = TransactionSessionManagerFactory.findGlobalTransactionSession(GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort));
		return globalTransactionSession.getGlobalTransactionStatus();
	}
//

	@Override
	public Long branchTransactionRegister(String resourceId, String clientIp, String globalTransactionIdWithAddressAndPort, String tableNameAndLockRecordIdsStringsString) throws GlobalTransactionException
	{
		GlobalTransactionSession globalTransactionSession = checkGlobalTransactionSessionOk(GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort), GlobalTransactionStatus.begin);

		BranchTransactionSession branchTransactionSession = new BranchTransactionSession();
		branchTransactionSession.setGlobalTransactionId(GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort));
		branchTransactionSession.setBranchTransactionId(TransactionIds.generateTransactionId());
		branchTransactionSession.setApplicationId(globalTransactionSession.getApplicationId());
		branchTransactionSession.setTransactionServiceGroup(globalTransactionSession.getTransactionServiceGroup());
		// branchTransactionSession.setBranchType(branchType);
		branchTransactionSession.setResourceId(resourceId);
		branchTransactionSession.setTableNameAndLockRecordIdsStringsString(tableNameAndLockRecordIdsStringsString);
		branchTransactionSession.setClientIp(clientIp);

		if (!branchTransactionSession.lock())
		{
			throw new GlobalTransactionException(TransactionExceptionCode.lock_conflict);
		}
		try
		{
			globalTransactionSession.addBranchTransactionSession(branchTransactionSession);
		} catch (RuntimeException ex)
		{
			throw new GlobalTransactionException(TransactionExceptionCode.failed_add_branch_transaction);

		}
		return branchTransactionSession.getBranchTransactionId();
	}

	@Override
	public void branchTransactionReport(String globalTransactionIdWithAddressAndPort, long branchTransactionId, BranchTransactionStatus branchTransactionStatus, String applicationData) throws GlobalTransactionException
	{
		GlobalTransactionSession globalTransactionSession = TransactionSessionManagerFactory.findGlobalTransactionSession(GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort));
		if (globalTransactionSession == null)
		{
			throw new GlobalTransactionException(TransactionExceptionCode.global_transaction_not_exist, "" + GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort) + "");
		}
		BranchTransactionSession branchTransactionSession = globalTransactionSession.getBranchTransactionSession(branchTransactionId);
		if (branchTransactionSession == null)
		{
			throw new GlobalTransactionException(TransactionExceptionCode.branch_transaction_not_exist);
		}
		globalTransactionSession.changeBranchTransactionStatus(branchTransactionSession, branchTransactionStatus);
	}

	// 下面是私有方法
	//
	private void queueToRetryCommit(GlobalTransactionSession globalTransactionSession) throws GlobalTransactionException
	{
		globalTransactionSession.addSessionLifecycleListener(TransactionSessionManagerFactory.getRetryCommitTransactionSessionManager());
		TransactionSessionManagerFactory.getRetryCommitTransactionSessionManager().addGlobalTransactionSession(globalTransactionSession);
		globalTransactionSession.changeGlobalTransactionStatus(GlobalTransactionStatus.commit_retrying);
	}

	private void queueToRetryRollback(GlobalTransactionSession globalTransactionSession) throws GlobalTransactionException
	{
		globalTransactionSession.addSessionLifecycleListener(TransactionSessionManagerFactory.getRetryRollbackTransactionSessionManager());
		TransactionSessionManagerFactory.getRetryRollbackTransactionSessionManager().addGlobalTransactionSession(globalTransactionSession);
		GlobalTransactionStatus globalTransactionStatus = globalTransactionSession.getGlobalTransactionStatus();
		if (globalTransactionStatus.name().startsWith("Timeout"))
		{
			globalTransactionSession.changeGlobalTransactionStatus(GlobalTransactionStatus.timeout_rollback_retrying);
		} else
		{
			globalTransactionSession.changeGlobalTransactionStatus(GlobalTransactionStatus.rollback_retrying);
		}
	}
	//

}
